package server

import (
	"dag2.0/gat1400-proxy/db"
	"dag2.0/gat1400-proxy/kafka"
	"dag2.0/gat1400-proxy/logger"
	"dag2.0/gat1400-proxy/model"
	"dag2.0/gat1400-proxy/model/constant"
	"dag2.0/gat1400-proxy/operator"
	dag_plugin_kafkaconsumer "dag2.0/gat1400-proxy/operator/e-kafkaconsumer"
	"dag2.0/gat1400-proxy/util/times"
	"errors"
	"fmt"
	jsoniter "github.com/json-iterator/go"
	"os"
	"strings"
)

const (
	OPERATION_ADD    = 1
	OPERATION_UPDATE = 3
	OPERATION_DELETE = 2

	RESOURCE_EQ = 2
	RESOURCE_TG = 3

	EXECUTE_OPERATION_ADD = 1
	EXECUTE_OPERATION_UPD = 2
	EXECUTE_OPERATION_DEL = 3
)

type UrmNotify struct {
	AlterTime     int    `json:"alterTime"`
	Id            string `json:"id"`
	OperationType int    `json:"operationType"`
	ResourceType  int    `json:"resourceType"`
}

// 消费urm通知kafka，收集资源变更情况
func watchResourceToNotify() {
	//启动kafka消费 urm变更通知
	urmNotifyKafka := os.Getenv("URM_NOTIFY_KAFKA")
	if len(urmNotifyKafka) == 0 {
		return
	}
	consumer := dag_plugin_kafkaconsumer.KafkaConsumer{}
	consumer.Init(operator.NewViperConfig(map[string]interface{}{
		"kafkaconsumer_bootstrap":          urmNotifyKafka,
		"kafkaconsumer_topics":             "urm_resource",
		"kafkaconsumer_groupId":            "dag_1400_watch",
		"kafkaconsumer_fromEarliestOffset": false,
		"kafkaconsumer_batchSize":          1000,
		"kafkaconsumer_batchDelay":         10000,
		"kafkaconsumer_parallel":           5,
	}), func(data interface{}) error {
		kafkaMsgs, ok := data.([]*kafka.KafkaMessage)
		if !ok {
			logger.LOG_WARN(fmt.Sprintf("Handle [kafkamsgtohubmsg] 数据格式错误，need []*kafka.KafkaMessage , get %T", data))
			return errors.New(fmt.Sprintf("Handle [kafkamsgtohubmsg] 数据格式错误，need []*kafka.KafkaMessage , get %T", data))
		}
		if len(kafkaMsgs) == 0 {
			return nil
		}
		//收集更新的设备id，卡口id
		var updEqIds = make(map[string]bool)
		var updTgIds = make(map[string]bool)
		var delEqIds = make(map[string]bool)
		var delTgIds = make(map[string]bool)

		for _, msg := range kafkaMsgs {
			un := &UrmNotify{}
			err := jsoniter.Unmarshal(msg.Value, un)
			if err != nil {
				logger.LOG_WARN("urm通知转换异常：", string(msg.Value), err)
				continue
			}
			switch un.OperationType {
			case OPERATION_UPDATE:
				if un.ResourceType == RESOURCE_EQ {
					updEqIds[un.Id] = true
				} else if un.ResourceType == RESOURCE_TG {
					updTgIds[un.Id] = true
				}
			case OPERATION_DELETE:
				if un.ResourceType == RESOURCE_EQ {
					delEqIds[un.Id] = true
					delete(updEqIds, un.Id)
				} else if un.ResourceType == RESOURCE_TG {
					delTgIds[un.Id] = true
					delete(updTgIds, un.Id)
				}
			}
		}
		if len(updEqIds) > 0 || len(updTgIds) > 0 || len(delEqIds) > 0 || len(delTgIds) > 0 {
			err := watchNotify(updEqIds,
				updTgIds,
				delEqIds,
				delTgIds)
			if err != nil {
				logger.LOG_WARN("设备变更通知异常：", err)
			}
		}
		return nil
	})
}

// 通知上级，资源的变更、删除
func watchNotify(updEqIds, updTgIds, delEqIds, delTgIds map[string]bool) error {
	//1.查询订阅及绑定信息
	subs, err := db.QuerySubscribes()
	if err != nil {
		return err
	}

	var activeManagerSubs = make(map[string][]*model.SubscribeLib)
	for _, sub := range subs {
		if sub.SubscribeStatus == 0 || sub.SubscribeStatus == 99 {
			detail := "," + sub.SubscribeDetail + ","
			if strings.Index(detail, ","+constant.EQUIPMENT_LIST+",") >= 0 ||
				strings.Index(detail, ","+constant.EQUIPMENT_STATUS+",") >= 0 ||
				strings.Index(detail, ","+constant.TOLLGATE_LIST+",") >= 0 ||
				strings.Index(detail, ","+constant.SINGLE_TOLLGATE_INFO+",") >= 0 ||
				strings.Index(detail, ","+constant.LAND_LIST+",") >= 0 ||
				strings.Index(detail, ","+constant.SINGLE_LAND_INFO+",") >= 0 ||
				strings.Index(detail, ","+constant.HIK_ILLEGAL+",") >= 0 {
				if _, ok := activeManagerSubs[sub.ManagerID]; !ok {
					activeManagerSubs[sub.ManagerID] = make([]*model.SubscribeLib, 0)
					activeManagerSubs[sub.ManagerID] = append(activeManagerSubs[sub.ManagerID], sub)
				}
			}
		}
	}
	if len(activeManagerSubs) == 0 {
		return nil
	}
	//2.查询活跃订阅的资源绑定,汇聚成用于通知的虚拟视图库对象
	activeManagerExs := make([]*model.SubscribeManagerEx, 0)
	for mId, _ := range activeManagerSubs {
		bindResources, err := db.QueryByViewId(mId)
		if err != nil || bindResources == nil {
			logger.LOG_WARN(err)
			continue
		}
		manager, err := db.QueryManagerById(mId)
		if err != nil {
			logger.LOG_WARN(err)
			continue
		}
		smexUpd := &model.SubscribeManagerEx{
			SubscribeManager: manager,
			ExecuteOperation: EXECUTE_OPERATION_UPD,
		}
		smexDel := &model.SubscribeManagerEx{
			SubscribeManager: manager,
			ExecuteOperation: EXECUTE_OPERATION_DEL,
		}
		for _, eqId := range bindResources.EquipmentIds {
			//改设备
			if _, ok := updEqIds[eqId]; ok {
				smexUpd.EquipmentIds = append(smexUpd.EquipmentIds, eqId)
			}
			//删设备
			if _, ok := delEqIds[eqId]; ok {
				smexDel.EquipmentIds = append(smexDel.EquipmentIds, eqId)
			}
		}
		for _, tgId := range bindResources.TollgateIds {
			//改卡口
			if _, ok := updTgIds[tgId]; ok {
				smexUpd.TollgateIds = append(smexUpd.TollgateIds, tgId)
			}
			//删卡口
			if _, ok := delTgIds[tgId]; ok {
				smexDel.TollgateIds = append(smexDel.TollgateIds, tgId)
			}
		}
		if len(smexUpd.EquipmentIds) > 0 || len(smexUpd.TollgateIds) > 0 {
			activeManagerExs = append(activeManagerExs, smexUpd)
		}
		if len(smexDel.EquipmentIds) > 0 || len(smexDel.TollgateIds) > 0 {
			activeManagerExs = append(activeManagerExs, smexDel)
		}
	}

	if len(activeManagerExs) == 0 {
		return nil
	}
	//3.通知
	for _, amex := range activeManagerExs {
		asubs := activeManagerSubs[amex.ManagerID]
		if len(asubs) == 0 {
			logger.LOG_WARN("未找到通知订阅对象:", amex.ManagerID)
			continue
		}
		for _, asub := range asubs {
			NotificationResource(&model.SubscribeLib{
				Subscribe:   asub.Subscribe,
				ManagerID:   asub.ManagerID,
				ManagerName: amex.ManagerName,
				UpdateTime:  times.Time2Str(),
				InsertTime:  times.Time2Str(),
			}, amex, false)
		}
	}

	return nil
}
